package com.zjbd.douyangcommoncore.utils;

import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.util.MimeTypeUtils;

import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Objects;

/**
 * RocketMQ 工具类（终极适配 2.3.4 版本，基于源码 1:1 对齐）
 * 支持：同步/异步/单向/延迟/事务/批量/顺序/广播消息
 * 关键说明：广播模式由消费者端配置，生产者无需设置
 */
@Slf4j
@Component
public class RocketMQUtil {

    private final RocketMQTemplate rocketMQTemplate;
    private final ObjectMapper objectMapper;

    // 注入源码原生 RocketMQTemplate + Spring 自动配置的 ObjectMapper
    public RocketMQUtil(RocketMQTemplate rocketMQTemplate, ObjectMapper objectMapper) {
        this.rocketMQTemplate = rocketMQTemplate;
        this.objectMapper = objectMapper;
    }

    // ==================== 通用返回结果（简洁易用）====================
    @Data
    @Builder
    @NoArgsConstructor
    @AllArgsConstructor
    public static class SendResultVO {
        private boolean success;    // 是否发送成功
        private String msgId;       // 消息ID（批量消息返回批次ID）
        private String sendStatus;  // 发送状态（SEND_OK/FLUSH_DISK_TIMEOUT 等）
        private String errMsg;      // 异常信息（失败时非空）

        // 快速创建成功结果
        public static SendResultVO success(SendResult sendResult) {
            return SendResultVO.builder()
                    .success(true)
                    .msgId(sendResult.getMsgId())
                    .sendStatus(sendResult.getSendStatus().name())
                    .build();
        }

        // 快速创建失败结果
        public static SendResultVO fail(String errMsg) {
            return SendResultVO.builder()
                    .success(false)
                    .errMsg(errMsg)
                    .build();
        }
    }

    // ==================== 基础配置常量（与源码默认值对齐）====================
    private static final long DEFAULT_TIMEOUT = 3000L;         // 默认超时时间（ms）
    private static final int MAX_DELAY_LEVEL = 18;             // 最大延迟级别（源码支持 1-18）
    private static final int MIN_DELAY_LEVEL = 1;              // 最小延迟级别

    // ==================== 1. 同步发送（最常用，源码重载对齐）====================
    /**
     * 极简同步发送（默认超时）
     * @param topic 主题（支持 tag：topic:tag）
     * @param data 消息内容（自动序列化）
     */
    public SendResultVO syncSend(String topic, Object data) {
        return syncSend(topic, data, DEFAULT_TIMEOUT);
    }

    /**
     * 带超时的同步发送
     * @param topic 主题
     * @param data 消息内容
     * @param timeout 超时时间（ms）
     */
    public SendResultVO syncSend(String topic, Object data, long timeout) {
        try {
            // 构建 Spring Message（与源码 createRocketMqMessage 逻辑一致）
            org.springframework.messaging.Message<Object> message = MessageBuilder
                    .withPayload(data)
                    .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON_VALUE)
                    .build();

            // 调用源码 syncSend 重载方法
            SendResult sendResult = rocketMQTemplate.syncSend(topic, message, timeout);
            log.info("同步发送消息成功 | topic: {}, msgId: {}", topic, sendResult.getMsgId());
            return SendResultVO.success(sendResult);
        } catch (Exception e) {
            String errMsg = String.format("同步发送消息失败 | topic: %s, 原因: %s", topic, e.getMessage());
            log.error(errMsg, e);
            return SendResultVO.fail(errMsg);
        }
    }

    // ==================== 2. 异步发送（源码重载对齐）====================
    /**
     * 极简异步发送（默认超时）
     * @param topic 主题
     * @param data 消息内容
     * @param callback 发送回调（成功/失败处理）
     */
    public void asyncSend(String topic, Object data, SendCallback callback) {
        asyncSend(topic, data, callback, DEFAULT_TIMEOUT);
    }

    /**
     * 带超时的异步发送
     * @param topic 主题
     * @param data 消息内容
     * @param callback 回调函数
     * @param timeout 超时时间（ms）
     */
    public void asyncSend(String topic, Object data, SendCallback callback, long timeout) {
        try {
            org.springframework.messaging.Message<Object> message = MessageBuilder
                    .withPayload(data)
                    .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON_VALUE)
                    .build();

            // 调用源码 asyncSend 重载方法
            rocketMQTemplate.asyncSend(topic, message, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    log.info("异步发送消息成功 | topic: {}, msgId: {}", topic, sendResult.getMsgId());
                    callback.onSuccess(sendResult);
                }

                @Override
                public void onException(Throwable e) {
                    String errMsg = String.format("异步发送消息失败 | topic: %s, 原因: %s", topic, e.getMessage());
                    log.error(errMsg, e);
                    callback.onException(e);
                }
            }, timeout);
        } catch (Exception e) {
            log.error("异步发送消息初始化失败 | topic: {}", topic, e);
            callback.onException(e);
        }
    }

    // ==================== 3. 单向发送（无返回，高吞吐）====================
    /**
     * 单向发送（适合日志、监控等非核心消息）
     */
    public void onewaySend(String topic, Object data) {
        try {
            org.springframework.messaging.Message<Object> message = MessageBuilder
                    .withPayload(data)
                    .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON_VALUE)
                    .build();

            // 调用源码 sendOneWay 方法
            rocketMQTemplate.sendOneWay(topic, message);
            log.info("单向发送消息成功 | topic: {}", topic);
        } catch (Exception e) {
            log.error("单向发送消息失败 | topic: {}", topic, e);
        }
    }

    // ==================== 4. 延迟消息发送（复用源码 delayLevel 重载）====================
    /**
     * 极简延迟消息（默认超时）
     * @param topic 主题
     * @param data 消息内容
     * @param delayLevel 延迟级别（1-18）：1=1s,2=5s,3=10s,4=30s,5=1min...18=2h
     */
    public SendResultVO syncSendDelay(String topic, Object data, int delayLevel) {
        return syncSendDelay(topic, data, delayLevel, DEFAULT_TIMEOUT);
    }

    /**
     * 完整参数延迟消息（直接调用源码重载，无需手动构建原生 Message）
     */
    public SendResultVO syncSendDelay(String topic, Object data, int delayLevel, long timeout) {
        try {
            // 验证延迟级别（与源码支持范围一致）
            if (delayLevel < MIN_DELAY_LEVEL || delayLevel > MAX_DELAY_LEVEL) {
                throw new IllegalArgumentException(
                        String.format("延迟级别必须在 %d-%d 之间", MIN_DELAY_LEVEL, MAX_DELAY_LEVEL));
            }

            org.springframework.messaging.Message<Object> message = MessageBuilder
                    .withPayload(data)
                    .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON_VALUE)
                    .build();

            // 关键：调用源码 syncSend(destination, message, timeout, delayLevel) 重载
            SendResult sendResult = rocketMQTemplate.syncSend(topic, message, timeout, delayLevel);
            log.info("延迟消息发送成功 | topic: {}, msgId: {}, 延迟级别: {}", topic, sendResult.getMsgId(), delayLevel);
            return SendResultVO.success(sendResult);
        } catch (Exception e) {
            String errMsg = String.format("延迟消息发送失败 | topic: %s, 延迟级别: %d, 原因: %s", topic, delayLevel, e.getMessage());
            log.error(errMsg, e);
            return SendResultVO.fail(errMsg);
        }
    }

    // ==================== 5. 事务消息发送（源码参数对齐）====================
    /**
     * 事务消息发送（监听器需通过 @RocketMQTransactionListener 注解声明）
     * @param topic 主题
     * @param data 消息内容
     * @param transactionArg 传递给本地事务的参数
     */
    public TransactionSendResult sendTransactionMessage(String topic, Object data, Object transactionArg) {
        try {
            org.springframework.messaging.Message<Object> message = MessageBuilder
                    .withPayload(data)
                    .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON_VALUE)
                    .build();

            // 调用源码 sendMessageInTransaction 方法（3个参数，监听器注解绑定）
            TransactionSendResult result = rocketMQTemplate.sendMessageInTransaction(topic, message, transactionArg);
            log.info("事务半消息发送成功 | topic: {}, msgId: {}, 事务状态: {}",
                    topic, result.getMsgId(), result.getLocalTransactionState());
            return result;
        } catch (Exception e) {
            String errMsg = String.format("事务消息发送失败 | topic: %s, 原因: %s", topic, e.getMessage());
            log.error(errMsg, e);
            throw new RuntimeException(errMsg, e);
        }
    }

    /**
     * 事务消息发送（无额外参数）
     */
    public TransactionSendResult sendTransactionMessage(String topic, Object data) {
        return sendTransactionMessage(topic, data, null);
    }

    // ==================== 6. 批量消息发送（复用源码批量方法）====================
    /**
     * 批量发送消息（直接调用源码 syncSend(Collection)，内部自动处理 MessageBatch）
     * @param topic 主题（支持 tag：topic:tag）
     * @param dataList 消息列表
     */
    public SendResultVO batchSend(String topic, List<Object> dataList) {
        return batchSend(topic, dataList, DEFAULT_TIMEOUT);
    }

    /**
     * 带超时的批量发送
     */
    public SendResultVO batchSend(String topic, List<Object> dataList, long timeout) {
        if (CollectionUtils.isEmpty(dataList)) {
            return SendResultVO.fail("批量消息列表不能为空");
        }

        try {
            // 构建 Spring Message 集合（与源码批量处理逻辑一致）
            Collection<org.springframework.messaging.Message<Object>> messageList = dataList.stream()
                    .map(data -> MessageBuilder
                            .withPayload(data)
                            .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON_VALUE)
                            .build())
                    .filter(Objects::nonNull)
                    .toList();

            // 调用源码批量发送方法（内部已处理 MessageBatch 拆分和大小校验）
            SendResult sendResult = rocketMQTemplate.syncSend(topic, messageList, timeout);
            log.info("批量消息发送成功 | topic: {}, 总条数: {}, msgId: {}", topic, dataList.size(), sendResult.getMsgId());
            return SendResultVO.success(sendResult);
        } catch (Exception e) {
            String errMsg = String.format("批量消息发送失败 | topic: %s, 总条数: %d, 原因: %s",
                    topic, dataList.size(), e.getMessage());
            log.error(errMsg, e);
            return SendResultVO.fail(errMsg);
        }
    }

    // ==================== 7. 顺序消息发送（源码重载对齐）====================
    /**
     * 顺序消息发送（同一 hashKey 消息有序）
     * @param topic 主题
     * @param data 消息内容
     * @param hashKey 哈希键（相同 key 发送到同一队列）
     */
    public SendResultVO orderSend(String topic, Object data, String hashKey) {
        return orderSend(topic, data, hashKey, DEFAULT_TIMEOUT);
    }

    /**
     * 带超时的顺序消息
     */
    public SendResultVO orderSend(String topic, Object data, String hashKey, long timeout) {
        try {
            org.springframework.messaging.Message<Object> message = MessageBuilder
                    .withPayload(data)
                    .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON_VALUE)
                    .build();

            // 调用源码 syncSendOrderly 重载方法
            SendResult sendResult = rocketMQTemplate.syncSendOrderly(topic, message, hashKey, timeout);
            log.info("顺序消息发送成功 | topic: {}, hashKey: {}, msgId: {}", topic, hashKey, sendResult.getMsgId());
            return SendResultVO.success(sendResult);
        } catch (Exception e) {
            String errMsg = String.format("顺序消息发送失败 | topic: %s, hashKey: %s, 原因: %s",
                    topic, hashKey, e.getMessage());
            log.error(errMsg, e);
            return SendResultVO.fail(errMsg);
        }
    }

    // ==================== 8. 广播消息发送（2.3.4 版本正确实现）====================
    /**
     * 广播消息发送（核心说明：广播模式由消费者端配置，生产者无需设置）
     * 消费者配置要求：@RocketMQMessageListener(messageModel = MessageModel.BROADCASTING)
     */
    public SendResultVO broadcastSend(String topic, Object data) {
        return broadcastSend(topic, data, DEFAULT_TIMEOUT);
    }

    /**
     * 带超时的广播消息（生产者端仅需普通发送，消费者端配置广播模式）
     */
    public SendResultVO broadcastSend(String topic, Object data, long timeout) {
        try {
            org.springframework.messaging.Message<Object> message = MessageBuilder
                    .withPayload(data)
                    .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON_VALUE)
                    .build();

            // 2.3.4 版本无 setMessageModel 方法！广播模式由消费者配置，生产者直接普通发送
            SendResult sendResult = rocketMQTemplate.syncSend(topic, message, timeout);
            log.info("广播消息发送成功（消费者需配置 BROADCASTING 模式）| topic: {}, msgId: {}", topic, sendResult.getMsgId());
            return SendResultVO.success(sendResult);
        } catch (Exception e) {
            String errMsg = String.format("广播消息发送失败 | topic: %s, 原因: %s", topic, e.getMessage());
            log.error(errMsg, e);
            return SendResultVO.fail(errMsg);
        }
    }

    // ==================== 扩展：带自定义属性的消息 ====================
    /**
     * 同步发送带自定义属性的消息（keys、用户属性等）
     * @param topic 主题
     * @param data 消息内容
     * @param keys 消息 keys（用于查询）
     * @param customProps 自定义属性
     */
    public SendResultVO syncSendWithProps(String topic, Object data, String keys, Map<String, Object> customProps) {
        try {
            MessageBuilder<Object> builder = MessageBuilder
                    .withPayload(data)
                    .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON_VALUE);

            // 设置 keys（与源码 KEYS 头一致）
            if (Objects.nonNull(keys)) {
                builder.setHeader("KEYS", keys);
            }

            // 设置自定义属性
            if (!CollectionUtils.isEmpty(customProps)) {
                customProps.forEach(builder::setHeader);
            }

            SendResult sendResult = rocketMQTemplate.syncSend(topic, builder.build(), DEFAULT_TIMEOUT);
            log.info("带属性消息发送成功 | topic: {}, keys: {}, msgId: {}", topic, keys, sendResult.getMsgId());
            return SendResultVO.success(sendResult);
        } catch (Exception e) {
            String errMsg = String.format("带属性消息发送失败 | topic: %s, keys: %s, 原因: %s",
                    topic, keys, e.getMessage());
            log.error(errMsg, e);
            return SendResultVO.fail(errMsg);
        }
    }

    // ==================== 工具方法：获取原生实例（如需扩展）====================
    public RocketMQTemplate getRocketMQTemplate() {
        return rocketMQTemplate;
    }

    public ObjectMapper getObjectMapper() {
        return objectMapper;
    }
}